package com.raymond.queue.handler;

import com.raymond.queue.Consumption;
import com.raymond.queue.FileQueue;
import com.raymond.queue.QueueManager;
import com.raymond.queue.common.msg.request.ConsumptionMsgRequest;
import com.raymond.queue.common.msg.response.ConsumptionMsgResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * 生产者请求处理
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-07-14 11:04
 */
public class ConsumptionMsgRequestHandler extends SimpleChannelInboundHandler<ConsumptionMsgRequest> {
    private final static Logger logger = LoggerFactory.getLogger(ConsumptionMsgRequestHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ConsumptionMsgRequest msg) {
        try {
            logger.info("消息消费,topic:{},group:{}", msg.getTopic(), msg.getGroupName());
            Consumption<byte[]> consumption = QueueManager.getFileQueue(msg.getTopic()).getConsumption(msg.getGroupName());
            if (consumption == null) {
                ctx.writeAndFlush(new ConsumptionMsgResponse(msg.getSequenceId(), 1));
            } else {
                ctx.writeAndFlush(new ConsumptionMsgResponse(msg.getSequenceId(), consumption.pollBytes(msg.getCount())));
            }
        } catch (Exception e) {
            ctx.writeAndFlush(new ConsumptionMsgResponse(msg.getSequenceId(), 1));
            logger.error("消息消费异常,客户端地址:{},topic:{},group:{}", ctx.channel().remoteAddress(), msg.getTopic(), msg.getGroupName(), e);
        }
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("ConsumptionMsgRequestHandler Established connection with the remote client.");

        // do something

        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("ConsumptionMsgRequestHandler Disconnected with the remote client.");

        // do something

        ctx.fireChannelInactive();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}
